package net.pantolomin.nayo;

import java.io.IOException;
import java.net.InetAddress;
import java.net.SocketAddress;
import java.net.SocketOption;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.AbstractSelectableChannel;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import net.pantolomin.nayo.api.Configuration;
import net.pantolomin.nayo.api.ConnectionEventsListener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The provider for NIO operations
 */
@SuppressWarnings("rawtypes")
public final class NioProvider {
	/** The provider's id. */
	private static final AtomicInteger providerId = new AtomicInteger(0);

	/** The logger */
	private final Logger logger;

	/** The provider's id. */
	private final int id;

	/** The provider's state */
	private final AtomicReference<State> state;

	/** The configuration */
	private final Configuration configuration;

	/** The thread group */
	private final ThreadGroup threadGroup;

	/** The connectors */
	private final Map<AbstractSelectableChannel, NioConnector<? extends AbstractSelectableChannel>> connectors;

	/** The contexts per channel */
	private final Map<SocketChannel, NioConnection> channelContexts;

	/** The selection loop */
	private NioSelectionLoop selectionLoop;

	/** The timer */
	private Timer timer;

	/** The executor for reading */
	private ExecutorService readExecutor;

	/** The executor for writing */
	private ExecutorService writeExecutor;

	/**
	 * Create the provider
	 * 
	 * @param configuration
	 *          the configuration
	 */
	public NioProvider(Configuration configuration) {
		this.logger = LoggerFactory.getLogger(getClass());
		this.id = providerId.incrementAndGet();
		this.state = new AtomicReference<State>(State.STOPPED);
		this.configuration = configuration;
		this.threadGroup = new ThreadGroup("NIO-" + this.id);
		this.connectors = new ConcurrentHashMap<AbstractSelectableChannel, NioConnector<? extends AbstractSelectableChannel>>();
		this.channelContexts = new IdentityHashMap<SocketChannel, NioConnection>();
	}

	/**
	 * @return the configuration
	 */
	public Configuration getConfiguration() {
		return this.configuration;
	}

	/**
	 * @return the selection loop
	 */
	NioSelectionLoop getSelectionLoop() {
		return this.selectionLoop;
	}

	/**
	 * @return the executor for writing
	 */
	ExecutorService getWriteExecutor() {
		return this.writeExecutor;
	}

	/**
	 * @return the timer for scheduled execution
	 */
	Timer getTimer() {
		return this.timer;
	}

	/**
	 * Start the provider
	 * 
	 * @throws IOException
	 *           if the selector could not be opened
	 */
	public void start() throws IOException {
		if (this.state.compareAndSet(State.STOPPED, State.STARTING)) {
			// Start the selector loop
			this.selectionLoop = new NioSelectionLoop(this);
			Thread selectionThread = new Thread(this.threadGroup, this.selectionLoop, "selection-loop");
			try {
				selectionThread.setPriority(selectionThread.getPriority() + 1);
			} catch (Exception e) {
				this.logger.warn("Failed to set the selection thread a slightly higher priority", e);
			}
			selectionThread.start();
			this.readExecutor = this.configuration.getExecutorForReading(this.threadGroup);
			this.writeExecutor = this.configuration.getExecutorForWriting(this.threadGroup);
			this.timer = new Timer("apf-cnx-live-check", true);
			this.state.set(State.STARTED);
		} else {
			throw new IllegalStateException("Provider is not stopped");
		}
	}

	/**
	 * Stop all connections and shutdown the provider
	 */
	public void stop() {
		if (this.state.compareAndSet(State.STARTED, State.STOPPING)) {
			this.timer.cancel();
			this.timer = null;

			// Stop the selector loop - it will in turn close all connections
			// see "closeAllConnections" below
			this.selectionLoop.stop();
		} else {
			throw new IllegalStateException("Provider is not started");
		}
	}

	//
	//
	//
	//
	//
	//
	//
	//
	//
	//
	// ************************************************************************
	// ************************************************************************
	// Create the client & server connectors
	// ************************************************************************
	// ************************************************************************

	/**
	 * Start listening for connections
	 * 
	 * @param hostAddress
	 *          the host's address
	 * @param port
	 *          the port to listen on
	 * @param socketOptions
	 *          the socket options
	 * @param eventsListener
	 *          the listener for connection events
	 * @return the server connector
	 * @throws IOException
	 *           if the communication channel could not be opened
	 */
	public NioServer listen(InetAddress hostAddress, int port, Map<SocketOption, Object> socketOptions, ConnectionEventsListener eventsListener)
			throws IOException {
		NioServer server = new NioServer(this, hostAddress, port, socketOptions, eventsListener);
		this.connectors.put(server.getChannel(), server);
		try {
			server.start();
		} catch (RuntimeException e) {
			this.connectors.remove(server.getChannel());
			server.close();
			throw e;
		}
		return server;
	}

	/**
	 * Initiate a connection
	 * 
	 * @param hostAddress
	 *          the host's address
	 * @param port
	 *          the port to listen on
	 * @param socketOptions
	 *          the socket options
	 * @param eventsListener
	 *          the listener for connection events
	 * @return the client connector
	 * @throws IOException
	 *           if the communication channel could not be opened
	 */
	public NioClient connect(InetAddress hostAddress, int port, Map<SocketOption, Object> socketOptions, ConnectionEventsListener eventsListener)
			throws IOException {
		NioClient client = new NioClient(this, hostAddress, port, socketOptions, eventsListener);
		this.connectors.put(client.getChannel(), client);
		try {
			client.start();
		} catch (RuntimeException e) {
			this.connectors.remove(client.getChannel());
			client.close();
			throw e;
		}
		return client;
	}

	// ************************************************************************
	// ************************************************************************
	// The events triggered by the selection loop's thread
	// ************************************************************************
	// ************************************************************************

	/**
	 * Declare a new socket channel. This method is called only by the "selection thread".
	 * 
	 * @param serverChannel
	 *          the server channel
	 * @param socketChannel
	 *          the socket channel
	 * @param selectionKey
	 *          the selection key
	 */
	void addSocketChannel(ServerSocketChannel serverChannel, SocketChannel socketChannel, SelectionKey selectionKey) {
		NioServer connector = (NioServer) this.connectors.get(serverChannel);
		if (connector == null) {
			// the provider was stopped
			selectionKey.cancel();
			try {
				socketChannel.close();
			} catch (IOException e) {
				this.logger.error("Failed to close socket channel", e);
			}
			return;
		}
		NioConnection connection = new NioConnection(connector, socketChannel, selectionKey);
		try {
			connection.scheduleCheckIdle(this.timer, this.configuration.getMaxIdleTime());
		} catch (NullPointerException e) {
			if (this.timer == null) {
				connection.doClose();
				return;
			}
			throw e;
		}
		this.channelContexts.put(socketChannel, connection);
		try {
			connector.getEventsListener().newConnection(connection);
		} catch (Exception e) {
			this.logger.error("Failed to handle the event \"new connection\"", e);
		}
	}

	/**
	 * Declare a new socket channel. This method is called only by the "selection thread".
	 * 
	 * @param socketChannel
	 *          the socket channel
	 * @param selectionKey
	 *          the selection key
	 */
	void addSocketChannel(SocketChannel socketChannel, SelectionKey selectionKey) {
		NioClient connector = (NioClient) this.connectors.get(socketChannel);
		if (connector == null) {
			// the provider was stopped
			selectionKey.cancel();
			try {
				socketChannel.close();
			} catch (IOException e) {
				this.logger.warn("Failed to close socket channel", e);
			}
			return;
		}
		connector.finishConnect();
		NioConnection connection = new NioConnection(connector, socketChannel, selectionKey);
		try {
			connection.scheduleCheckIdle(this.timer, this.configuration.getMaxIdleTime());
		} catch (NullPointerException e) {
			if (this.timer == null) {
				this.connectors.remove(socketChannel);
				connection.doClose();
				return;
			}
			throw e;
		}
		this.channelContexts.put(socketChannel, connection);
		try {
			connector.getEventsListener().newConnection(connection);
		} catch (Exception e) {
			this.logger.error("Failed to handle the event \"new connection\"", e);
		}
	}

	/**
	 * Indicates that the remote peer closed a connection. This method is called only by the "selection thread".
	 * 
	 * @param channel
	 *          the channel
	 * @param cleanly
	 *          indicates if the connection was closed cleanly
	 */
	void closeConnectionByRemote(SocketChannel channel, boolean cleanly) {
		// Remove the client connector (will do nothing if it is a server side
		// connection)
		NioConnector<? extends AbstractSelectableChannel> connector = this.connectors.remove(channel);

		// Remove the connection and close it
		NioConnection cnx = this.channelContexts.remove(channel);
		if (cnx != null) {
			cnx.doClose();
			cnx.getConnector().getEventsListener().closedByRemote(cnx, cleanly);
		} else {
			SocketAddress remoteAddr;
			try {
				remoteAddr = channel.getRemoteAddress();
			} catch (IOException e1) {
				// never mind, log won't show what connection it was
				remoteAddr = null;
			}
			this.logger.error("Closing an unknown channel: {}", remoteAddr);
			try {
				channel.close();
			} catch (IOException e) {
				this.logger.warn("Failed to close the connection: {}", remoteAddr, e);
			}
		}

		if (connector != null) {
			// A client connector => close it also
			try {
				connector.close();
			} catch (IOException e) {
				this.logger.warn("Failed to close the connector: {}:{}", connector.getHostAddress(), Integer.valueOf(connector.getPort()), e);
			}
		}
	}

	/**
	 * Close a channel. This method is called only by the "selection thread".
	 * 
	 * @param channel
	 *          the channel
	 */
	void closeChannel(AbstractSelectableChannel channel) {
		NioConnector<? extends AbstractSelectableChannel> connector = this.connectors.remove(channel);
		if (connector != null) {
			// Either:
			// - a client connector, both the connector and the connection are
			// stopped
			// - the server connector, the connector and all attached
			// connections are stopped
			closeConnector(connector);
		} else {
			// A server connection (only close the connection, not the
			// connector)
			NioConnection cnx = this.channelContexts.remove(channel);
			if (cnx != null) {
				cnx.doClose();
				cnx.getConnector().getEventsListener().connectionClosed(cnx);
			} else {
				SocketChannel sChannel = (SocketChannel) channel;
				SocketAddress remoteAddr;
				try {
					remoteAddr = sChannel.getRemoteAddress();
				} catch (IOException e1) {
					// never mind, log won't show what connection it was
					remoteAddr = null;
				}
				this.logger.error("Closing an unknown channel: {}", remoteAddr);
				try {
					sChannel.close();
				} catch (IOException e) {
					this.logger.warn("Failed to close the connection: {}", remoteAddr, e);
				}
			}
		}
	}

	/**
	 * Close all connections. This method is called only by the "selection thread".
	 */
	void closeAllConnections() {
		// Go through all connectors and close them
		Iterator<NioConnector<? extends AbstractSelectableChannel>> itCnx = this.connectors.values().iterator();
		while (itCnx.hasNext()) {
			NioConnector<? extends AbstractSelectableChannel> connector = itCnx.next();
			itCnx.remove();
			closeConnector(connector);
		}

		// Finish stopping the provider
		this.writeExecutor.shutdown();
		this.readExecutor.shutdown();
		this.selectionLoop = null;
		this.state.set(State.STOPPED);
	}

	/**
	 * Close a connector and all its connections. This method is called only by the "selection thread".
	 * 
	 * @param connector
	 *          the connector
	 */
	private void closeConnector(NioConnector<? extends AbstractSelectableChannel> connector) {
		NioConnection cnx = this.channelContexts.remove(connector.getChannel());
		if (cnx != null) {
			// A client connector
			cnx.doClose();
			connector.getEventsListener().connectionClosed(cnx);
		} else {
			// A server connector => need to look for its connections
			Iterator<Entry<SocketChannel, NioConnection>> itCnx = this.channelContexts.entrySet().iterator();
			while (itCnx.hasNext()) {
				Entry<SocketChannel, NioConnection> entry = itCnx.next();
				cnx = entry.getValue();
				if (cnx.getConnector() == connector) {
					// Gotcha: remove and close it
					itCnx.remove();
					cnx.doClose();
					connector.getEventsListener().connectionClosed(cnx);
				}
			}
		}

		// Close the connector
		try {
			connector.close();
		} catch (IOException e) {
			this.logger.warn("Failed to close the connector: {}:{}", connector.getHostAddress(), Integer.valueOf(connector.getPort()), e);
		}
	}

	/**
	 * Indicates that the server was bound. This method is called only by the "selection thread".
	 * 
	 * @param socketChannel
	 *          the socket channel
	 */
	void bound(ServerSocketChannel socketChannel) {
		NioServer connector = (NioServer) this.connectors.get(socketChannel);
		connector.getEventsListener().bindDone(connector.getHostAddress(), connector.getPort());
	}

	/**
	 * Indicates that we failed to bind. This method is called only by the "selection thread".
	 * 
	 * @param socketChannel
	 *          the socket channel
	 * @param exception
	 *          the problem that occurred
	 */
	void failedToBind(ServerSocketChannel socketChannel, Exception exception) {
		NioServer connector = (NioServer) this.connectors.remove(socketChannel);
		if (connector != null) {
			try {
				connector.close();
				connector.getEventsListener().bindFailed(connector.getHostAddress(), connector.getPort(), exception.getMessage());
			} catch (Exception e) {
				this.logger.error("Failed to handle the event \"failed bind\"", e);
			}
		}
		// else: the provider was stopped
	}

	/**
	 * Indicates that we failed to connect. This method is called only by the "selection thread".
	 * 
	 * @param socketChannel
	 *          the socket channel
	 * @param exception
	 *          the problem that occurred
	 */
	void failedToConnect(SocketChannel socketChannel, Exception exception) {
		NioClient connector = (NioClient) this.connectors.remove(socketChannel);
		if (connector != null) {
			try {
				connector.close();
				connector.getEventsListener().connectionFailed(connector.getHostAddress(), connector.getPort(), exception.getMessage());
			} catch (Exception e) {
				this.logger.error("Failed to handle the event \"failed connection\"", e);
			}
		}
		// else: the provider was stopped
	}

	/**
	 * Indicates that we failed to connect in time. This method is called only by the "selection thread".
	 * 
	 * @param socketChannel
	 *          the socket channel
	 */
	void connectionTimeout(SocketChannel socketChannel) {
		NioClient connector = (NioClient) this.connectors.remove(socketChannel);
		if (connector != null) {
			try {
				connector.close();
				connector.getEventsListener().connectionTimeout(connector.getHostAddress(), connector.getPort());
			} catch (Exception e) {
				this.logger.error("Failed to handle the event \"connection timeout\"", e);
			}
		}
		// else: the provider was stopped
	}

	/**
	 * Read some data. This method is called only by the "selection thread".
	 * 
	 * @param socketChannel
	 *          the socket channel
	 */
	void read(SocketChannel socketChannel) {
		NioConnection connection = this.channelContexts.get(socketChannel);
		if (socketChannel.isConnected()) {
			this.readExecutor.execute(connection.getReader());
		} else {
			// The connection was cut
			connection.doClose();
			connection.getConnector().getEventsListener().closedByRemote(connection, false);
		}
	}

	/**
	 * Write some waiting data. This method is called only by the "selection thread".
	 * 
	 * @param socketChannel
	 *          the socket channel
	 */
	void write(SocketChannel socketChannel) {
		NioConnection connection = this.channelContexts.get(socketChannel);
		if (socketChannel.isConnected()) {
			connection.getWriter().resumeWriting(this.writeExecutor);
		} else {
			// The connection was cut
			connection.doClose();
			connection.getConnector().getEventsListener().closedByRemote(connection, false);
		}
	}

	/**
	 * The state of the provider
	 */
	private static enum State {
		/** Stopped */
		STOPPED,
		/** Starting */
		STARTING,
		/** Started */
		STARTED,
		/** Stopping */
		STOPPING
	}
}